﻿using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.ActiveMQ.Commands;
using Infrastructure;
using Nest;
using System;
using System.Threading.Tasks;

namespace Infrastructure.MQ
{
    /// <summary>
    /// 消息队列辅助类
    /// </summary>
    public static class MQHelper
    {
        /// <summary>
        /// 连接字符串
        /// </summary>
        private static string ConnectionString = Config.ReadSetting("MQ");

        /// <summary>
        /// 异步发送消息
        /// </summary>
        /// <param name="topicChannel">主题通道</param>
        /// <param name="filterKey">过滤器键名</param>
        /// <param name="filterValue">过滤器键值</param>
        /// <param name="content">文本消息内容</param>
        /// <param name="ip">消息队列服务IP</param>
        /// <param name="port">消息队列服务端口</param>
        public static async Task<bool> SendAsync(string topicChannel, string filterKey, string filterValue, string content, string ip = null, int port = 61616)
        {
            // 检查配置
            if (string.IsNullOrEmpty(ip))
            {
                if (string.IsNullOrEmpty(ConnectionString))
                    throw new Exception("appsettings.json does not define mq url section:[MQ]");
            }

            // 异步执行发送
            return await Task.Factory.StartNew(() =>
            {
                return Send(topicChannel, filterKey, filterValue, content, ip, port);
            });
        }

        /// <summary>
        /// 同步发送消息
        /// </summary>
        /// <param name="topicChannel">主题通道</param>
        /// <param name="filterKey">过滤器键名</param>
        /// <param name="filterValue">过滤器键值</param>
        /// <param name="content">文本消息内容</param>
        /// <param name="ip">消息队列服务IP</param>
        /// <param name="port">消息队列服务端口</param>
        public static bool Send(string topicChannel, string filterKey, string filterValue, string content, string ip = null, int port = 61616)
        {
            // 连接字符串
            string connectionString = null;

            // 检查配置
            if (string.IsNullOrEmpty(ip))
            {
                if (string.IsNullOrEmpty(ConnectionString))
                    throw new Exception("appsettings.json does not define mq url section:[MQ]");

                connectionString = ConnectionString;
            }
            else
            {
                connectionString = $"tcp://{ip}:{port}/";
            }

            // 执行发送
            try
            {
                // 创建连接工厂
                IConnectionFactory factory = new ConnectionFactory(connectionString);

                using (IConnection connection = factory.CreateConnection())
                {
                    // 创建会话
                    using (ISession session = connection.CreateSession())
                    {
                        // 创建生产者，主题传入参数确定
                        IMessageProducer producer = session.CreateProducer(new ActiveMQTopic(topicChannel));

                        // 准备消息
                        ITextMessage msg = producer.CreateTextMessage();
                        msg.Text = content;

                        // 设置消息属性，用于消费者订阅时的消息过滤
                        // 该属性与消费者端的Selector对应，只有消费者设置的selector与消息的Properties匹配，
                        // 消息才会发给该消费者
                        if (!string.IsNullOrEmpty(filterKey))
                        {
                            msg.Properties.SetString(filterKey, filterValue);
                        }

                        // 发送消息
                        producer.Send(msg, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);

                        // 关闭
                        producer.Close();

                        return true;
                    }
                }
            }
            catch (Exception ex)
            {
#pragma warning disable CS4014 // 由于此调用不会等待，因此在调用完成前将继续执行当前方法
                LogHelper.Error("Send To ActiveMQ Failed.", ex);
#pragma warning restore CS4014 // 由于此调用不会等待，因此在调用完成前将继续执行当前方法
                return false;
            }
        }

        /// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="topicChannel">主题通道</param>
        /// <param name="clientId">客户端Id，用于持久订阅用，如果不需要持久订阅可以传空</param>
        /// <param name="selector">消息过滤器，与发布消息的filterKey相关，如果接收所有消息则可以传空</param>
        /// <param name="action">接收消息后执行的自定义动作，其中string类型的参数为接收到的文本消息内容</param>
        /// <param name="connectionString">消息队列地址，默认从配置文件获取，如果传入则以传入的为准</param>
        public static void Receive(string topicChannel, string clientId, string selector, Action<string> action
        , string connectionString = null)
        {
            // 如果未传入连接地址，则从配置文件获取
            if (string.IsNullOrEmpty(connectionString))
            {
                // 检查配置
                if (string.IsNullOrEmpty(ConnectionString))
                    throw new Exception("appsettings.json does not define mq url section:[MQ]");

                connectionString = ConnectionString;
            }

            // 创建连接工厂
            IConnectionFactory factory = new ConnectionFactory(connectionString);

            // 创建连接
            IConnection connection = factory.CreateConnection();

            // 设置客户端Id，用于持久订阅
            if (!string.IsNullOrEmpty(clientId))
            {
                connection.ClientId = clientId;
            }

            // 连接到ActiveMQ
            connection.Start();

            // 创建会话
            ISession session = connection.CreateSession();

            // 创建消费者，订阅传入的主题
            // 可以设置selector指定订阅的消息属性
            // 需参考生产者设置的消息属性
            // 接收所有消息则设置为null
            IMessageConsumer consumer = session.CreateDurableConsumer(new ActiveMQTopic(topicChannel)
                , clientId
                , selector
                , false);

            // 设置接收消息的监听器，执行对应的自定义操作
            consumer.Listener += new MessageListener((message) =>
            {
                //将收到的消息按文本消息进行转换
                ITextMessage msg = (ITextMessage)message;

                //执行传入的自定义操作
                action(msg.Text);
            });
        }
    }
}
